package david.java.flink_sql;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.types.Row;
import scala.Tuple3;

/**
 * @Description:
 * @Author: David
 * @Date: Create in 9:14 下午 2022/1/12
 */
public class Tuple3TableSource implements StreamTableSource<Row> {

    private DataStream<Tuple3<String, Long, Long>> dataStream;
    private DataStream<Row> rowStream;

    public Tuple3TableSource(DataStream<Tuple3<String, Long, Long>> dataStream) {
        this.dataStream = dataStream;
    }

    @Override
    public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
        return rowStream;
    }

    @Override
    public TableSchema getTableSchema() {
        String[] names = {"name", "value", "timeStamp"};
        DataType[] types = {DataTypes.STRING(), DataTypes.BIGINT(), DataTypes.BIGINT()};

        return TableSchema.builder().fields(names, types).build();
    }


    public static class Tuple3ToRowFunction implements MapFunction<Tuple3<String, Long, Long>, Row>, ResultTypeQueryable<Row> {

        private String[] fieldNames;
        private DataType[] dataTypes;
        private TypeInformation[] typeInformations;

        public Tuple3ToRowFunction(String[] fieldNames, DataType[] dataTypes) {
            this.fieldNames = fieldNames;
            this.dataTypes = dataTypes;
            this.typeInformations = new TypeInformation[fieldNames.length];

            for (int i = 0; i < fieldNames.length; i++) {
                typeInformations[i] = TypeConversions.fromDataTypeToLegacyInfo(dataTypes[i]);
            }

        }

        @Override
        public Row map(Tuple3<String, Long, Long> value) throws Exception {
            Row row = new Row(3);
            row.setField(0, value._1());
            row.setField(1, value._2());
            row.setField(2, value._3());
            return row;
        }

        @Override
        public TypeInformation<Row> getProducedType() {
            return new RowTypeInfo(typeInformations,fieldNames);
        }
    }

}
